@Leondon9 This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Merge conflicts: This PR has merge conflicts with main and cannot be merged as-is. Rebase locally and resolve them: git fetch upstream main && git rebase upstream/main, resolve the conflicts, then force-push.

Note: Your branch is 648 commits behind main. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.

What to do next:

• Fix each issue listed above.
• Make sure static checks pass locally (prek run --from-ref main --stage pre-commit).
• Mark the PR as "Ready for review" when you're done.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.

Hi @Leondon9,

After merging #64610 and #64611, my intention was to create a generic sensor for asset events rather than one based only on partitions. It would support filtering by partition, event metadata (extra), or event time.

I've opened #70225 and am testing it now. Since it is more generic and does not require changes to Airflow core, I think it has a better chance of being merged than this PR.

Could you check whether it addresses the needs you described in this discussion?

Thanks for the heads-up @hussein-awala, and for landing #64610/#64611 — this PR now sits directly on
top of them. I also agree on the destination: a generic AssetEventSensor filtering on extra, time
range, expected_count and process_result is a superset of what this PR does, and it's the better
long-term shape.

On the core dependency, one clarification. That dependency was always meant to be temporary — the PR
body (June 3) said the lookup commit should be dropped or reduced once #64610 landed, and I've now
done that: the core example Dag moved into the provider's own example_dags/, and what remains under
airflow-core/ is 10 lines in triggerer_job_runner.py plus a regression test.

That said, I don't think those 10 lines can go away while keeping deferrable mode, and I think #70225
needs them too. A trigger has neither a metadata-DB session nor an Execution API client, so its only
route to asset events is SUPERVISOR_COMMS, and that route goes through core:

• InletEventsAccessor sends GetAssetEventByAsset (sdk/execution_time/context.py), and
  TriggerCommsDecoder.send() bridges both the event-loop thread (greenback) and worker threads
  (run_coroutine_threadsafe) — so the sync_to_async(...) wrapper in Add AssetEventSensor to standard provider #70225 does reach the
  supervisor rather than short-circuiting earlier.
• The supervisor validates that frame against the ToTriggerSupervisor discriminated union
  (decoder.validate_python in supervisor.py). On main that union has no GetAssetEventByAsset
  member, so the frame fails with pydantic union_tag_invalid.
• Even past validation, the dispatch chain in TriggerRunnerSupervisor._handle_request ends in
  raise ValueError(f"Unknown message type {type(msg)}").

I walked into this myself: I had added the dispatch branches but missed the union entry, so my own
deferred path was broken, and my trigger tests didn't catch it because they replace SUPERVISOR_COMMS
with a mock. The DB-backed SUPERVISOR_COMMS stand-in in #70225's conftest.py substitutes the same
channel, so the suite can stay green while the deferred path would still hit the union check above.
This PR now adds the union entry plus a test asserting both union membership and the dispatch branch —
deleting the two union lines makes it fail with that error. Might be worth confirming on your side
before #70225 lands; happy to be wrong about it.

So I'd read the core change as a prerequisite rather than a difference in scope: whichever sensor
lands, deferrable asset-event lookups need that wiring.

On how to proceed — I don't think two PRs touching the same four provider files helps reviewers.
Either direction works for me:

1. Keep Add asset partition sensor #67941 as the base (open since June 3, currently conflict-free, now with a how-to guide and a
   provider example Dag) and fold the generalisation from Add AssetEventSensor to standard provider #70225 into it, with you as co-author; or
2. Close this in favour of Add AssetEventSensor to standard provider #70225, with the triggerer wiring and its regression test carried over.

I genuinely have no preference between those and will follow whatever a maintainer prefers. My only
concern is that the wiring doesn't get dropped along the way, since without it deferrable=True fails
at runtime while the test suite stays green.

On your question: yes, a generic sensor covers what I described in #67375, with one requirement
carried over from it. The sketch there was partition_key="{{ data_interval_start | ds }}" with
deferrable=True, and question 4 in that discussion was precisely about matching semantics — "latest
event exists, any event exists, source-run-aware matching". That is what after settles here: with no
time bound, an event carrying a reused key from an earlier interval satisfies the wait immediately.
For date-shaped keys that is harmless; for keys that repeat across runs (a region code) it is wrong.
So as long as #70225 keeps both partition_key and a time bound templated per run, it covers my case.